package com.yao.system.mq;


import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping("/rocket")
@RequiredArgsConstructor
public class RocketMqController {

    Logger log = LoggerFactory.getLogger(RocketMqController.class);

    private final RocketMQTemplate rocketMQTemplate;

    @GetMapping("/send")
    public void sendMessage(String topic, String msg) {
        //构造消息
        Message<String> message = MessageBuilder.withPayload("hello").build();
        //1.普通发送消息
        rocketMQTemplate.send("test2", message);
        //2.同步发送消息，会阻塞，得到结果
        SendResult result = rocketMQTemplate.syncSend("test2", message);
        //3.异步发送消息，有回调
        rocketMQTemplate.asyncSend("test2", message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("发送失败:{}", throwable);
            }
        });
        //4.广播消息，所有订阅topic的消费者都能收到消息，默认为集群模式只有单个消费者实例能够收到消息
        Message<String> message1 = MessageBuilder.withPayload(msg)
                .setHeader("MSG_TYPE", "Broadcasting") // 设置广播模式
                .build();
        rocketMQTemplate.send(topic, message1);
        //5.消息过滤，指定tag
        Message<String> message2 = MessageBuilder.withPayload(msg)
                .setHeader("tag", "tag1")
                .build();
        rocketMQTemplate.send(topic, message2);
        //6.顺序消息
        Message<String> message3 = MessageBuilder.withPayload(msg)
                .setHeader("orderId", "1")
                .build();
        //RocketMQHeaders.TAGS;
        try {
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, message3, "1");
        } catch (Exception e) {
            e.printStackTrace();
        }
        //7.延时消息
        Message<String> message4 = MessageBuilder.withPayload(msg)
                .build();
        rocketMQTemplate.syncSendDelayTimeSeconds(topic, message4, 5000L);
        //8.事务消息
        rocketMQTemplate.sendMessageInTransaction("test-transaction", message, null);
    }
}
